package io.github.bigbird0101.datatransfer.container;

import cn.hutool.json.JSONUtil;
import integration.modle.exception.IntegrationValidException;
import integration.utils.trace.LogUtils;
import io.github.bigbird0101.datatransfer.collector.Communication;
import io.github.bigbird0101.datatransfer.collector.CommunicationTool;
import io.github.bigbird0101.datatransfer.constants.State;
import lombok.extern.slf4j.Slf4j;

import java.text.SimpleDateFormat;

/**
 * 阻塞任务容器
 */
@Slf4j
public class TaskContainer extends AbstractContainer {

    private static final SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    /*
    1、获取到页面配置
    -------
    2、创建一个任务容器
    3、配置转换
    4、读写端类型
    5、执行运行前置操作
    6、创建读写端
    7、数据转换
    8、运行
     */
    TaskExecutor taskExecutor;
    private long startTimeStamp;

    private long endTimeStamp;

    private int totalStage = 1;

    private String taskInstanceId;


    public TaskContainer(String config) {
        super(config);
        // TODO: 任务实例id
        taskInstanceId = JSONUtil.parseObj(config).getStr("taskId");
        taskExecutor = new TaskExecutor(config);
        taskExecutor.init();
    }


    public void start() {
        try {
            this.startTimeStamp = System.currentTimeMillis();


            taskExecutor.start();

            /**
             * 状态check时间间隔
             */
            int sleepIntervalInMillSec = 100;

            long reportIntervalInMillSec = 2000;

            long lastReportTimeStamp = 0;

            Communication lastTaskGroupContainerCommunication = new Communication();

            boolean failedOrKilled = false;
            while (true) {

                Communication taskCommunication = taskExecutor.getTaskCommunication();

                // 如果当前时间已经超出汇报时间的interval，那么我们需要马上汇报
                long now = System.currentTimeMillis();
                if (now - lastReportTimeStamp > reportIntervalInMillSec) {
                    lastTaskGroupContainerCommunication = reportTaskGroupCommunication(lastTaskGroupContainerCommunication, 1);
                    lastReportTimeStamp = now;
                }
                Thread.sleep(sleepIntervalInMillSec);

                if (!taskCommunication.isFinished()) {
                    continue;
                }
                if (taskCommunication.getState() == State.FAILED) {
                    failedOrKilled = true;
                    break;
                } else if (taskCommunication.getState() == State.KILLED) {
                    failedOrKilled = true;
                    break;
                } else if (taskExecutor.isTaskFinished()) {
                    break;
                }

            }
            // 2.发现该taskExecutor的状态失败则汇报错误
            if (failedOrKilled) {
                lastTaskGroupContainerCommunication = reportTaskGroupCommunication(lastTaskGroupContainerCommunication, 1);
                throw new IntegrationValidException("运行时出错", lastTaskGroupContainerCommunication.getThrowable());
            }

            //最后汇报一次
            reportTaskGroupCommunication(lastTaskGroupContainerCommunication, 1);
//            LogUtils.info(log,CommunicationTool.Stringify.getSnapshot(taskExecutor.getTaskCommunication()));

        } catch (Throwable e) {
            throw new IntegrationValidException("任务运行异常", e);
        } finally {
            this.endTimeStamp = System.currentTimeMillis();
            // 日志汇报
            this.logStatistics();
        }
    }

    public void sendData(String data) {
        LogUtils.warn(log, "nothing");
        throw new IntegrationValidException("不支持此方法");
    }

    /**
     * TODO：模拟运行中停止
     */
    public void startAndStop(int time) {
        try {
            this.startTimeStamp = System.currentTimeMillis();

            taskExecutor.init();
            taskExecutor.start();

            /**
             * 状态check时间间隔
             */
            int sleepIntervalInMillSec = 100;

            long reportIntervalInMillSec = 2000;

            long lastReportTimeStamp = 0;

            Communication lastTaskGroupContainerCommunication = new Communication();

            boolean failedOrKilled = false;
            while (true) {

                Communication taskCommunication = taskExecutor.getTaskCommunication();

                Thread.sleep(sleepIntervalInMillSec);
                // 如果当前时间已经超出汇报时间的interval，那么我们需要马上汇报
                long now = System.currentTimeMillis();
                if (now - lastReportTimeStamp > reportIntervalInMillSec) {
                    lastTaskGroupContainerCommunication = reportTaskGroupCommunication(lastTaskGroupContainerCommunication, 1);
                    lastReportTimeStamp = now;
                }
                if (System.currentTimeMillis() - this.startTimeStamp > time) {
                    stop();
                }
                if (!taskCommunication.isFinished()) {
                    continue;
                }
                if (taskCommunication.getState() == State.FAILED) {
                    failedOrKilled = true;
                    break;
                } else if (taskCommunication.getState() == State.KILLED) {
                    failedOrKilled = true;
                    break;
                } else if (taskExecutor.isTaskFinished()) {
                    break;
                }

            }
            // 2.发现该taskExecutor的状态失败则汇报错误
            if (failedOrKilled) {
                lastTaskGroupContainerCommunication = reportTaskGroupCommunication(lastTaskGroupContainerCommunication, 1);
                throw new IntegrationValidException("运行时出错", lastTaskGroupContainerCommunication.getThrowable());
            }

            //最后汇报一次
            reportTaskGroupCommunication(lastTaskGroupContainerCommunication, 1);
//            LogUtils.info(log,CommunicationTool.Stringify.getSnapshot(taskExecutor.getTaskCommunication()));

        } catch (Throwable e) {
            throw new IntegrationValidException("任务运行异常", e);
        } finally {
            this.endTimeStamp = System.currentTimeMillis();
            // 日志汇报
            this.logStatistics();
        }
    }

    public void stop() {
        taskExecutor.stop();
    }

    public void shutdown() {
        taskExecutor.shutdown();
    }

    private Communication reportTaskGroupCommunication(Communication lastTaskGroupContainerCommunication, int taskCount) {
        long currentTimeStamp = System.currentTimeMillis();
        long totalCosts = (currentTimeStamp - this.startTimeStamp) / 1000;
        if (totalCosts == 0L) {
            totalCosts = 1L;
        }
        Communication taskCommunication = taskExecutor.getTaskCommunication();
        taskCommunication.setTimestamp(System.currentTimeMillis());
        Communication reportCommunication = CommunicationTool.getReportCommunication(taskCommunication,
                lastTaskGroupContainerCommunication, taskCount);

        // 字节速率
        long byteSpeedPerSecond = taskCommunication.getLongCounter(CommunicationTool.READ_SUCCEED_BYTES)
                / totalCosts;

        long recordSpeedPerSecond = taskCommunication.getLongCounter(CommunicationTool.READ_SUCCEED_RECORDS)
                / totalCosts;

        lastTaskGroupContainerCommunication.setLongCounter(CommunicationTool.BYTE_SPEED, byteSpeedPerSecond);
        lastTaskGroupContainerCommunication.setLongCounter(CommunicationTool.RECORD_SPEED, recordSpeedPerSecond);

        LogUtils.info(log, CommunicationTool.Stringify.getSnapshot(taskExecutor.getTaskCommunication()));
        return reportCommunication;
    }

    private void logStatistics() {
        long totalCosts = (this.endTimeStamp - this.startTimeStamp) / 1000;
        if (totalCosts == 0L) {
            totalCosts = 1L;
        }
        Communication communication = this.taskExecutor.getTaskCommunication();
        communication.setTimestamp(this.endTimeStamp);

        Communication tempComm = new Communication();
        tempComm.setTimestamp(this.endTimeStamp);

        Communication reportCommunication = CommunicationTool.getReportCommunication(communication, tempComm, this.totalStage);

        // 字节速率
        long byteSpeedPerSecond = communication.getLongCounter(CommunicationTool.READ_SUCCEED_BYTES)
                / totalCosts;

        long recordSpeedPerSecond = communication.getLongCounter(CommunicationTool.READ_SUCCEED_RECORDS)
                / totalCosts;

        reportCommunication.setLongCounter(CommunicationTool.BYTE_SPEED, byteSpeedPerSecond);
        reportCommunication.setLongCounter(CommunicationTool.RECORD_SPEED, recordSpeedPerSecond);


        LogUtils.info(log, String.format(
                "\n任务ID:" + taskInstanceId +
                        "\n" + "%-26s: %-18s\n" + "%-26s: %-18s\n" + "%-26s: %19s\n"
                        + "%-26s: %19s\n" + "%-26s: %19s\n" + "%-26s: %19s\n"
                        + "%-26s: %19s\n",
                "任务启动时刻",
                dateFormat.format(startTimeStamp),

                "任务结束时刻",
                dateFormat.format(endTimeStamp),

                "任务总计耗时",
                String.valueOf(totalCosts) + "s",
                "任务平均流量",
                CommunicationTool.Stringify.stringify(byteSpeedPerSecond)
                        + "/s",
                "记录写入速度",
                String.valueOf(recordSpeedPerSecond)
                        + "rec/s", "读出记录总数",
                String.valueOf(CommunicationTool.getTotalReadRecords(communication)),
                "读写失败总数",
                String.valueOf(CommunicationTool.getTotalErrorRecords(communication))
        ));

        if (communication.getLongCounter(CommunicationTool.TRANSFORMER_SUCCEED_RECORDS) > 0
                || communication.getLongCounter(CommunicationTool.TRANSFORMER_FAILED_RECORDS) > 0
                || communication.getLongCounter(CommunicationTool.TRANSFORMER_FILTER_RECORDS) > 0) {
            LogUtils.info(log, String.format(
                    "\n" + "%-26s: %19s\n" + "%-26s: %19s\n" + "%-26s: %19s\n",
                    "Transformer成功记录总数",
                    communication.getLongCounter(CommunicationTool.TRANSFORMER_SUCCEED_RECORDS),

                    "Transformer失败记录总数",
                    communication.getLongCounter(CommunicationTool.TRANSFORMER_FAILED_RECORDS),

                    "Transformer过滤记录总数",
                    communication.getLongCounter(CommunicationTool.TRANSFORMER_FILTER_RECORDS)
            ));
        }


    }
}
